package com.shujia.flink.sink

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink

object Demo3FileSInk {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val sink = new BucketingSink[String]("flink/data/sink")


    //设置文件大小（默认文件大小是384MB）
    sink.setBatchSize(1024)

    //设置文件滚动周期，单位是毫秒（默认滚动周期是 Long.MAX_VALUE）
    sink.setBatchRolloverInterval( 10 * 1000)


    linesDS.addSink(sink)


    env.execute()

  }

}
